package profile.tagcombine2

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import profile.utils.TagCombineUtil

import scala.collection.mutable

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/23 16:26
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description: 当日多来源多模块标签大合并
  **/
object TagCombiner2 {

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)

    // 移动公司流量日志cm_log用户画像标签
    val cmTagPath = "G:\\data_shark\\testdata\\usertags\\out-cmlogtags2"
    // dsp请求日志dsp_log用户画像标签
    val dspTagPath = "G:\\data_shark\\testdata\\usertags\\out-dsptags2"
    // idmapping打通结果
    val idmpPath = "G:\\data_shark\\testdata\\usertags\\idmapping"

    // T日标签聚合结果输出目录
    val outpath = "G:\\data_shark\\testdata\\usertags\\out-combined-tags\\T日"

    val spark = SparkSession.builder().appName("当日标签合并").master("local").getOrCreate()
    import spark.implicits._

    /**
      * 加载数据
      */
    val cmTags: DataFrame = spark.read.parquet(cmTagPath)
    val dspTags: DataFrame = spark.read.parquet(dspTagPath)

    // 加载idmapping数据，并以map形式收集到driver端
    val idmp = spark.read.textFile(idmpPath)

    cmTags.show(10, false)
    dspTags.show(10,false)
    idmp.show(10,false)



    /**
      * 为cmtags数据查询追加统一图id（gid）
      */
    // 加工idmapping数据，成为kv结构（gid,idList）然后收集到driver端变成hashmap
    val idmpMap: collection.Map[String, List[String]] = idmp.map(line => {
      val split = line.split("\001")
      val gid = split(0)
      val ids = split(1).split(",")
      (gid, ids.toList)
    }).rdd.collectAsMap()

    // 将idmapping数据的kv调换，从guid->idsList  转换成: id->guid
    // 方便逆向查询： 从id查所属的gid
    val idmap = new mutable.HashMap[String, String]()
    for ((k, v) <- idmpMap) {
      for (id <- v) {
        idmap.put(id, k)
      }
    }

    // 将整理好的idmapping数据广播出去
    val bc = spark.sparkContext.broadcast(idmap)


    // 为cmtags查询追加统一图id
    val cmTagsWithGid = cmTags.rdd
      .map(row => {
        val phone = row.getAs[String]("phone")
        val idsTags = row.getAs[Map[String,Map[String,Double]]]("idsTags")
        val deviceTags = row.getAs[Map[String, Map[String, Double]]](2)
        val goodsKwTags = row.getAs[Map[String, Map[String, Double]]](3)

        // 查询追加统一图id --> gid
        // 查询逻辑：从标签数据中得到各id，然后跟idmapping中的所有id求交集
        // 如果交集不为空，则说明能查到，且用交集中的任意一个id逆向查询获得统一图id
        val imeis: Map[String, Double] = idsTags.getOrElse("imei", Map.empty[String,Double])
        val imeiSet = imeis.keySet.map(_.hashCode.toString)

        // 取出广播变量中的idmapping字典
        val idmap = bc.value
        // idmapping字典中的所有标识码
        val idset = idmap.keySet


        // 将行中的标识码和idmapping字典中的标识码求交集
        val intersectId = idset.intersect(imeiSet)
        val gid = if(intersectId.size>0) idmap.getOrElse(intersectId.toList(0),phone) else phone

        // 返回追加图id的结果
        (gid,idsTags,deviceTags, goodsKwTags)
      })


    // 将dsp标签dataframe转成rdd，方便后续的reduce聚合
    val dspTagsWithGid = dspTags.rdd.map(row=>{
      val gid = row.getAs[String]("guid")
      val idsTags = row.getAs[Map[String,Map[String, Double]]]("idsTags")
      val dspKwTags = row.getAs[Map[String,Map[String, Double]]]("dspKwTags")
      (gid,idsTags,dspKwTags)
    })

    // 根据统一图id来join两个rdd
    val joinRdd = cmTagsWithGid.map(tp=>(tp._1,tp)).fullOuterJoin(dspTagsWithGid.map(tp=>(tp._1,tp)))
    joinRdd.cache()
    println("--------joinrdd------------------------------")
    joinRdd.take(10).foreach(println)

    //sys.exit(1)

    // joinRdd的结构： (gid,(Option((gid,idsMap,deviceTags,goodKwTags)),Option(gid,idsTags,dspKwTags)))
    // 过滤出关联不上的cmtags
    val cmOnlyRdd = joinRdd.filter(_._2._2.isEmpty).map(_._2._1.get)

    // 过滤出关联不上的dsptags
    val dspOnlyRdd = joinRdd.filter(_._2._1.isEmpty).map(_._2._2.get)

    // 过滤出能关联上的cmtags和dsptags
    val toMergeRdd = joinRdd.filter(tp=>(!tp._2._1.isEmpty && !tp._2._2.isEmpty))

    // 对能关联上的标签rdd需要进行相同标签模块（测试数据中为idsMap标签模块）合并
    // toMergedRdd结构:(gid,(some((gid,idsMap,deviceTags,goodKwTags)),some(gid,idsTags,dspKwTags)))
    val mergedTags = toMergeRdd.map(tp => {
      val gid = tp._1
      //获取cmTags，结构： 1.gid, 2.idsTags, 3.deviceTags, 4.goodsKwTags
      val cmTags = tp._2._1.get
      //获取dspTags，结构：1.guid, 2.idsTags, 3.dspKwTags
      val dspTags = tp._2._2.get

      // 合并idsTags==> 逻辑结构： 合并两个Map[String,Double]
      val mergedIdsTags: Map[String, Map[String, Double]] = TagCombineUtil.mergeTagMap(cmTags._2, dspTags._2)

      // 封装最终结果
      TagCombineBean2(gid, mergedIdsTags, cmTags._3, cmTags._4, dspTags._3)
    })
    //println("--------需要多方合并的标签模块的合并结果------------------------------")
    //mergedTags.toDS().show(10,false)

    // 将cmOnlyrdd变换成统一格式  RDD[TagMergeBean2]
    val cmonlyTags = cmOnlyRdd.map(tp => {
      TagCombineBean2(tp._1, tp._2, tp._3, tp._4, Map.empty)
    })


    // dspOnlyRdd  RDD[TagMergeBean2]
    val dsponlyTags = dspOnlyRdd.map(tp => {
      TagCombineBean2(tp._1, tp._2, Map.empty, Map.empty,tp._3)
    })

    // 将3块标签数据union合并得到最终combine结果
    val resTags = mergedTags.union(cmonlyTags).union(dsponlyTags).toDS()
    resTags.show(10,false)

    resTags.coalesce(1).write.parquet(outpath)

    spark.close()
  }

}
